#include "pj_manager.hpp"
#include "json/json.h"

/*************************************   CONSTANTS  *************************************/
static const int TCPPORT = 8070;
static const int MANAGER_WAIT_WORK_EXIT_INTERVALTIME = 100000;   //manager process wait for worker exit interval time
static const int MANAGER_WAIT_WORK_EXIT_MAX_LOOP = 600;   //manager process wait for worker exit loop times
static const int MANAGER_WAIT_WORK_INIT_INTERVALTIME = 100000;   //manager process wait for worker init interval time
static const int MANAGER_WAIT_WORK_INIT_MAX_LOOP = 50;   //manager process wait for worker init loop times
//note: must make sure that MANAGER_WAIT_WORK_EXIT_MAX_LOOP >> NUM_OF_WORKERS
static const int NUM_OF_WORKERS = 2;

static const int RECV_TOTAL_MAXSIZE = 409600;

static const int RECV_GROUP_MAXSIZE = 1400;
static const int SEND_GROUP_MAXSIZE = 450;

/**************************************  STATIC VARIALBES *******************************/
static int gblListenFd;
static vector<pid_t> workersPids(NUM_OF_WORKERS, 0);
static vector<pid_t> workersAcceptFd(NUM_OF_WORKERS, 0);
static vector<bool> workersState(NUM_OF_WORKERS, false);  //false:free, true: busy

/************************************** SHARED VARIABLES ********************************/
static vector<sem_t*> sharedSems(NUM_OF_WORKERS);  //warning: must init manually them before use!
static vector<struct_val_manager_worker*> sharedMMaps(NUM_OF_WORKERS, nullptr);
static sem_t* shared_par_man;
static struct_val_parent_manager* shared_parent_svpm;

/************************************** MODULE FUNCTIONS ********************************/
/*
 * parse data for sending
 * para: const unordered_map<string, string>& um
 * return: string
 *
 * this function execs: unordered_map<string, string>->JSON string
 */
string parseSendData(const unordered_map<string, string>& um){
    Json::Value v;
    
    for (const auto& iterum:um) v[iterum.first] = iterum.second;
    
    Json::StreamWriterBuilder sw;
    sw.settings_["indentation"] = "";
    std::ostringstream osStream;
    string resStr;

    Json::StreamWriter* psw = sw.newStreamWriter();
    psw->write(v, &osStream);
    return osStream.str();
}

/*
 * parse data of recving
 * in default, it is an unordered_map
 * para: const string& raw, Json::Value& root
 * return pair<bool, Json::Value>: code true with none, false with error string
 *
 * note: set value of root inside the function
 */
pair<bool, string> parseRecvData(const string& raw, Json::Value& root){
    JSONCPP_STRING errs;
    Json::CharReaderBuilder readerBuilder;

    std::unique_ptr<Json::CharReader> const jsonReader(readerBuilder.newCharReader());
    bool res;
    res = jsonReader->parse(raw.c_str(), raw.c_str() + raw.length(), &root, &errs);
    if (!res || !errs.empty()) return {false, (string)errs};
    else return {true, ""};
}

/*
 * recv data, with recv() and send() inside, recv grouped data
 * para: int connfd
 * return pair<bool, string>: code true with recv content, false with error content
 */
pair<bool, string> recvData(int connfd){
    string res;
    while(true){
        char buff[RECV_GROUP_MAXSIZE + 1];
        memset(buff, 0, sizeof(char)*(RECV_GROUP_MAXSIZE + 1));
    recvData_innerstart:
        int n = recv(connfd, buff, RECV_GROUP_MAXSIZE, 0);
        if (n < 0){
            if (errno == EINTR || errno == EWOULDBLOCK || errno == EAGAIN) {    //not fatal error
                cout << "recvData group step, recv interrupt: " << strerror(errno) << "(errno: " << errno << ")" << endl;
                goto recvData_innerstart;
            }
            else{
                cout << "recvData group step, recv error: " << strerror(errno) << "(errno: " << errno << ")" << endl;
                return {false, ""};
            }
        }
        else{
            buff[n] = '\0';
            string recvGroupRecvdata = string(buff);
            if (n == RECV_GROUP_MAXSIZE){
                cout << "recvData group step, recv error: data too long!, data is:" << recvGroupRecvdata << endl;
                return {false, ""};
            }
            Json::Value rGR_v;
            pair<bool, string> rGR_ParseRet = parseRecvData(recvGroupRecvdata, rGR_v);
            if (!rGR_ParseRet.first || !rGR_v.isObject() || !rGR_v.isMember("t") || !rGR_v["t"].isString() ||
                !((rGR_v["t"].asString() == "0") || (rGR_v["t"].asString() == "1")) ||
                !rGR_v.isMember("l") || !rGR_v["l"].isString() || !rGR_v.isMember("d") || !rGR_v["d"].isString()){
                cout << "recvData group step, recv error: data format invalid, data is:" << recvGroupRecvdata << endl;
                return {false, ""};
            }
            else{
                int rGR_len;
                try{
                    rGR_len = stoi(rGR_v["l"].asString());
                }
                catch(...){
                    cout << "recvData group step, recv error: data format invalid, data is:" << recvGroupRecvdata << endl;
                    return {false, ""};
                }
                string realRecvStr = rGR_v["d"].asString();
                
                if (rGR_len != (int)realRecvStr.size()){
                    cout << "recvData group step, recv error: data format invalid, len check error, data is:" << recvGroupRecvdata << endl;
                    return {false, ""};
                }
                else{
                    res.append(realRecvStr.begin(), realRecvStr.end());
                    if (rGR_v["t"] == "1") break;
                    else if (res.size() > RECV_TOTAL_MAXSIZE){
                        cout << "recvData group step, recv error: data too long, data size is:" << recvGroupRecvdata.size() << endl;
                        return {false, ""};
                    }
                    else{
						cout << "recvData group step, recv ok: size is:" << n << endl;
                        //send 0(get group data ok) back
                        string respGroupSenddata = "{\"t\":\"0\"}";
                        if( send(connfd, respGroupSenddata.c_str(), respGroupSenddata.size(), 0) < 0){
                            cout << "recvData group step, send res data back error: " << strerror(errno) << "(errno: " << errno << ")" << endl;
                            return {false, ""};
                        }
                        continue;
                    }
                }
            }
        }
    }
	cout << endl;
    cout << "recvData: total size is:" << res.size() << endl;
    //cout << "recvData: recv msg from client:" << res << endl;
    return pair<bool, string>{true, res};
}

/*
 * parse and send data, with send() and recv() inside, send grouped data
 * para: const string& sendMsg, int connfd
 * return true if send ok else error
 */
bool sendData(const string& sendMsg, int connfd)
{
    int sMSize = sendMsg.size();
    cout << "LOG::client: send data total size:" << sMSize << endl;
    
    //拆分发送
    int sendIndex = 0;
    bool isLastGroup = false;
    while(sendIndex < sMSize){
        //gen send data1
        unordered_map<string, string> groupum;
        if ((sendIndex + SEND_GROUP_MAXSIZE) < sMSize){
            groupum["t"] = "0";
            groupum["l"] = to_string(SEND_GROUP_MAXSIZE);
            groupum["d"] = sendMsg.substr(sendIndex, SEND_GROUP_MAXSIZE);
        }
        else{
			isLastGroup = true;
            groupum["t"] = "1";
            groupum["l"] = to_string(sMSize - sendIndex);
            groupum["d"] = sendMsg.substr(sendIndex, sMSize - sendIndex);
        }
        //parse send data
        string groupStr = parseSendData(groupum);

        if( send(connfd, groupStr.c_str(), strlen(groupStr.c_str()), 0) < 0){
            cout << "sendData group step, send error: " << strerror(errno) << "(errno: " << errno << ")" << endl;
            return false;
        }
        else{
			if (isLastGroup){
				cout << "sendData group has sent all data" << endl;
				break;
			}
			
			//recv data, check it
			cout << "sendData group send one ok, size:" << " (" << groupum["l"] << "," << groupStr.size() << ") " << endl;
		
            char buff[RECV_GROUP_MAXSIZE + 1];
            memset(buff, 0, sizeof(char)*(RECV_GROUP_MAXSIZE + 1));
        sendData_innerstart:
            int n = recv(connfd, buff, RECV_GROUP_MAXSIZE, 0);
            if (n < 0){
                if (errno == EINTR || errno == EWOULDBLOCK || errno == EAGAIN) {//not fatal error
                    cout << "sendData group step, recv interrupt: " << strerror(errno) << "(errno: " << errno << ")" << endl;
                    goto sendData_innerstart;
                }
                else{
                    cout << "sendData group step, recv error: " << strerror(errno) << "(errno: " << errno << ")" << endl;
                    return false;
                }
            }
            else{
                buff[n] = '\0';
                string sendGroupRecvdata = string(buff);
                if (n == RECV_GROUP_MAXSIZE){
                    cout << "sendData group step, recv error: data too long!, data is:" << sendGroupRecvdata << endl;
                    return false;
                }
                Json::Value sGR_v;
                pair<bool, string> sGR_ParseRet = parseRecvData(sendGroupRecvdata, sGR_v);
                if (!sGR_ParseRet.first || !sGR_v.isObject() || !sGR_v.isMember("t") || !sGR_v["t"].isString() ||
                    sGR_v["t"].asString() != "0"
                    ){
                    cout << "sendData group step, recv error: data format invalid or server rep:false, data is:" << sendGroupRecvdata << endl;
                    return false;
                }
                else{
					cout << "sendData group recv one resp ok, ";
				    cout << "sendData group has some remaining data to send" << endl;
                    sendIndex += SEND_GROUP_MAXSIZE;
                    continue;
                }
            }
        }
    }
	cout << endl;
    return true;
}

/************************************ TOP LEVEL FUNCTIONS *******************************/
/*
 * manager shutdown
 * para: none, please use static gbl vals
 * return: none
 *
 * called by normal SIGUSR1 and shutdown = true
 */
void managerProcessShutDown(){
    //send SIGKILL to these workers
    for (int i = 0; i < NUM_OF_WORKERS; ++i){
        if (workersPids[i] != 0){
            if (kill(workersPids[i], SIGKILL) == -1){
                cout << "manager: " << getpid() << " send kill worker: " << workersPids[i] << "error" << endl;
            }
            else{
                cout << "manager: " << getpid() << " send kill worker: " << workersPids[i] << "success" << endl;
            }
        }
        else{
            cout << "manager: " << getpid() << " try killing worker, but find its id is:" << workersPids[i] << endl;
        }
    }
    
    for (int i = 0; i < NUM_OF_WORKERS; ++i){
        string semName = "/bla_psem_mmap" + to_string(i);
        if (sem_unlink(semName.c_str()) == -1){
            cerr << "sem_unlink" << endl;
        }
        if (munmap(sharedMMaps[i], sizeof(struct_val_manager_worker)) != 0){
            cerr << "munmap " << strerror(errno) << "(errno:" << errno << ")" << endl;
        }
    }
}

/*
 * close TCP server of manager process
 * para: none, please use static gbl vals
 * return: none
 */
void managerTCPclose(){
    close(gblListenFd);
}

/*
 * manager control
 * para: none, please use static gbl vals
 * return: none
 */
void managerProcessControl(){
    //display infos
    for (int i = 0; i < NUM_OF_WORKERS; ++i){
        if (workersState[i]) cout << "\t worker" << workersPids[i] << "is busy!" << endl;
        else cout << "\t worker" << workersPids[i] << "is free!" << endl;
    }
    cout << endl;
}

/*
 * signal handler of manager process, (special function)
 * para: int signo: ID of signal; use some static gbl vals inside
 * return: none
 *
 * note: may exit inside
 * note: has send() inside, and there is send()'s only location
 */
void managerSignal_handler(int signo){
    if (signo == SIGUSR1){
        cout << "SIGUSR1 of manager "<< getpid() << " start to manage!" << endl;
        sem_wait(shared_par_man);
        if (shared_parent_svpm->isShutdown){ //FORCE kill all producer and consumer processes
            sem_post(shared_par_man);
            cout << "manager "<< getpid() << " get the isShutdown signal" << endl;
            managerProcessShutDown();
        }
        else{
            sem_post(shared_par_man);
            managerProcessControl();
        }
    }
    else if (signo == SIGUSR2){
        cout << "SIGUSR2 of manager "<< getpid() << " ,should send something by TCP!" << endl;

        //scan all workers, find which should be send!
        for (int i = 0; i < NUM_OF_WORKERS; ++i){   //should be modified later, don't need scan all workers
            //sem_trywait, if can, means worker post the lock(task finished)
            int tempSem = sem_trywait(sharedSems[i]);
            if (tempSem == 0){  //success, automatically lock it again
                //call TCP send, use the memory of mem
                string sendMsg = sharedMMaps[i]->content;
                
                //call group send function
                if (!sendData(sendMsg, workersAcceptFd[i]))
                    cout << "sendData(group send data) has an internal error, please check it" << endl;
                
                //no matter sendData return true or false, close this acceptfd
                close(workersAcceptFd[i]);
                workersState[i] = false;
            }
            else if (errno == EAGAIN){    //the worker is waiting for task, or it is busy
                cout << "manager :"<< getpid() << " find that worker " << workersPids[i] << " isbusy" << endl;
            }
            else{  //something error happeded
                cout << "manager find that worker" << workersPids[i] << ", sem_trywait has unusual error" << strerror(errno) << " (errno:)" << errno << ")" << endl;
                continue;
            }
        }
    }
    else if (signo == SIGCHLD){
        for (int i = 0; i < MANAGER_WAIT_WORK_EXIT_MAX_LOOP; ++i) //note: if many workers exit in one loop, just one of them will be catched
        {
            usleep(MANAGER_WAIT_WORK_EXIT_INTERVALTIME);
            int childRetCode;
            int childRetPid = waitpid(-1, &childRetCode, WNOHANG);
            if (childRetPid == -1) {
                cout << "manager " << getpid() << " all worker have exited, so strange" << endl;
                break;
            } else if (childRetPid != 0){
                cout << "manager " << getpid() << " find that worker " << childRetPid << "has exited";
                cout << " WIFEXITED is:" << WIFEXITED(childRetCode) << " WEXITSTATUS is:" << WEXITSTATUS(childRetCode) << endl;
                //delete it of workersPids
                for (int nw = 0; nw < NUM_OF_WORKERS; ++nw){
                    if (workersPids[nw] == childRetPid){
                        workersPids[nw] = 0;
                    }
                }
            }else{   //some worker still work!
                continue;
            }
        }
        //checkpoint: if all worker exit, kill myself:)
        int remainWorkers = 0;
        for (int i = 0; i < NUM_OF_WORKERS; ++i){
            if (workersPids[i] != 0){
                remainWorkers++;
            }
        }
        
        if (remainWorkers == 0){
            cout << "manager " << getpid() << " all worker have exited, remainWorkers = 0" << endl;
            managerTCPclose();
            exit(0);  //kill itself
        }
        else{
            cout << "manager " << getpid() << " still wait for some workers exit" << endl;
        }
    }
    else{
        cout << "manager:" << getpid() << " unknown signo:" << signo << endl;
    }
}

/*
 * wait connect of TCP in manager process
 * commonly: manager exec: listen; worker exec:recv/send/close
 * server busy: manager exec: listen/recv/send/close
 * para: none, please use static gbl vals
 * return: none
 
 * note: in some fuck condition, abort()!
 */
void managerWaitConnect(){      
    int connfd = accept(gblListenFd, (struct sockaddr*)NULL, NULL);
    if (connfd == -1){
        if (errno == EINTR)
        {
            cout << "manager " << getpid() << " accept socket error: EINTR" << endl;
            return;
        }
        else{
            cout << "manager " << getpid() << " accept socket error: " << strerror(errno) << "(errno: " << errno << ")" << endl;
            return;
        }
    }
    else{
        //call recv group function
        cout << "manager " << getpid() << "accept success" << " connfd:" << connfd << endl;
        pair<bool, string> recvRet = recvData(connfd);
        
        if (recvRet.first){  //recv ok
            string recvStr = recvRet.second;
            cout << "LOG::manager: " << getpid() << " recv data length:" << recvStr.size() << endl;
            bool findFreeWorker = false;
            for (int i = 0; i < NUM_OF_WORKERS; ++i){
                if (!workersState[i]){
                    findFreeWorker = true;
                    workersState[i] = true;
                    workersAcceptFd[i] = connfd;
                    
                    memset(sharedMMaps[i]->content, 0, MMAP_CONTENT_LENGTH - 1);
                    const char* tempRecvStrC = recvStr.c_str();
                    memcpy(sharedMMaps[i]->content, tempRecvStrC, recvStr.size() + 1);
                    
                    if (kill(workersPids[i], SIGUSR2) == -1){
                        cout << "manager: " << getpid() << " send siguser2 worker: " << workersPids[i] << "error" << endl;
                    }
                    else{
                        cout << "manager: " << getpid() << " send siguser2 worker: " << workersPids[i] << "success" << endl;
                    }
                    findFreeWorker = true;
                    break;
                }
            }
            
            if (!findFreeWorker){
                //send server busy
                string sendMsg = "{\"code\": \"1\", \"info\": \"server busy\"}";
                //call group send function
                if(!sendData(sendMsg, connfd))
                    cout << "sendData error: " << strerror(errno) << "(errno: " << errno << ")" << endl;
                //no matter sendData return true or false, close this acceptfd
                close(connfd);
            }
        }
        else{  //recv err, directedly close this connfd
            close(connfd);
        }
    }
}

/*
 * init TCP server of manager process
 * para: int* ts: tcp server listen Fd
 * return: bool if ok true else false
 * 
 */
bool initTCPserver(int* ts){
    struct sockaddr_in servaddr;
    if( ((*ts) = socket(AF_INET, SOCK_STREAM, 0)) == -1 ){
        cout << "manager: " << getpid() << " create socket error: " << strerror(errno) << "(errno:" << errno<< ")" << endl;
        return false;
    }

    memset(&servaddr, 0, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    servaddr.sin_port = htons(TCPPORT);

    if( bind((*ts), (struct sockaddr*)&servaddr, sizeof(servaddr)) == -1){
        cout << "manager: " << getpid() << " bind socket error: " << strerror(errno) << "(errno:" << errno<< ")" << endl;
        return false;
    }
    if( listen((*ts), 10) == -1){
        cout << "manager: " << getpid() << " listen socket error: " << strerror(errno) << "(errno:" << errno<< ")" << endl;
        return false;
    }
    return true;
}

/*
 * manager TCP server
 * para: none, please use static gbl vals
 * return: none
 *
 * note: in some fuck condition, abort()!
 */
void managerTCPserver(){
    int TCPListenFd;
    
    if (!initTCPserver(&TCPListenFd))
    {
        cout << "manager " << getpid() << ": cannot init TCPserver!" << endl;
        abort();
    }
    else cout << "manager " << getpid() << ": init TCPserver success!" << endl;

    gblListenFd = TCPListenFd;
    while(1){
        //cout << "before managerWaitConnect" << endl;
        managerWaitConnect();
        //cout << "after managerWaitConnect" << endl;
    }
}

/*
 * process of manager, head function, with exit() inside, manage workers
 * para: sem_t* sem_par_man, struct_val_parent_manager* svpm
 */
void ManagerProcess(sem_t* sem_par_man, struct_val_parent_manager* svpm){
    shared_par_man = sem_par_man;
    shared_parent_svpm = svpm;
    cout << "manager " << getpid() << ": started" << endl;
    
    //set sems, mmaps and fork, note: from here, I will not call about() or exit
    //if you want to shutdown manager, send special signal to parent(then parent will send something to manager), don't use SIGKILL directedly!

    for (int i = 0; i < NUM_OF_WORKERS; ++i){
        //clear sems!!!
        string semName = "/bla_psem_mmap" + to_string(i);
        if (sem_unlink(semName.c_str()) == -1) cout << "manager: " << semName << endl;
        else cout << "manager: " << semName << " has been deleted" << endl;
        //sems, struct_sems init
        sharedSems[i] = sem_open(semName.c_str(), O_CREAT, 0644, 1);  //exclusive signal
        if (sharedSems[i] == SEM_FAILED){
            cerr << "sem_open" << endl;
        }
        
        //mmaps, struct_workers init
        string memtxtName = "temp_mem_txt" + to_string(i) + ".txt";
        int fd_vpm = open(memtxtName.c_str(), O_CREAT|O_RDWR|O_TRUNC, 0644);
        if (fd_vpm < 0) {
            cerr << "open" << endl;
        }
        unlink(memtxtName.c_str());
        if (ftruncate(fd_vpm, sizeof(struct_val_manager_worker)) == -1){
            cerr << "ftruncate" << endl;
        }
        struct_val_manager_worker* temp_svmw = (struct_val_manager_worker* )mmap(NULL, sizeof(struct_val_manager_worker), PROT_READ|PROT_WRITE, MAP_SHARED, fd_vpm, 0);
        if (temp_svmw == MAP_FAILED)
        {
            cerr << "mmap" << endl;
        }
        sharedMMaps[i] = temp_svmw;
        close(fd_vpm);
        
        //init state
        workersState[i] = false;
        
        //fork
        int workerid = fork();
        if (workerid == -1){
            cerr << "manager: fork" << endl;
            continue;
        }
        else if(workerid == 0){  //worker
            //call exit inside
            WorkerInterface(sharedSems[i], temp_svmw);
            exit(99);      //dump
        }else{  //parent
            workersPids[i] = workerid;
            continue;
        }
    }

    //wait for all worker finish initialized
    int temploopTime = 0;
    while(true){
        if (temploopTime > MANAGER_WAIT_WORK_INIT_MAX_LOOP){  //what fuck, cannot init all worker process
            cout << "not all worker init failed" << endl;
            managerProcessShutDown();
            exit(99);  //fuck!!!
        }
        bool waitOK = true;
        usleep(MANAGER_WAIT_WORK_INIT_INTERVALTIME);
        for (int i = 0; i < NUM_OF_WORKERS; ++i){
            //sem_trywait, if not, means worker has wait its lock(init OK)
            int tempSem = sem_trywait(sharedSems[i]);
            if (tempSem == 0){  //success, this worker hasn't inited, wait again
                cout << "manager find worker" << workersPids[i] << "not finied initializing" << endl;
                sem_post(sharedSems[i]);
                waitOK = false;
                break;
            }
            else if (errno == EAGAIN){    //the worker has inited
                continue;
            }
            else{  //something error happeded
                cout << "manager find that worker" << workersPids[i] << ", sem_trywait has unusual error" << strerror(errno) << " (errno:)" << errno << ")" << endl;
                waitOK = false;
                break;
            }
        }
        if (waitOK){
            break;
        }
    }
    cout << "now all worker init successfully" << endl;
    
    //add signals, must put it after fork workers!, otherwise you should overwrite signalhandler in workers
    struct sigaction newsignal;
    newsignal.sa_flags = 0; //this is a FUCK THING that musts be clear, C don't initialized memory with 0 automatically!
    newsignal.sa_handler = managerSignal_handler;
    
    //set signal handlers
    if (sigemptyset(&newsignal.sa_mask) == -1){
        cerr << "sigemptyset" << endl;
        abort();
    }

    if (sigaddset(&newsignal.sa_mask, SIGUSR1)){
        cerr << "sigaddset SIGUSR1" << endl;
        abort();
    }
    if (sigaddset(&newsignal.sa_mask, SIGUSR2)){
        cerr << "sigaddset SIGUSR2" << endl;
        abort();
    }
    if (sigaddset(&newsignal.sa_mask, SIGCHLD)){
        cerr << "sigaddset SIGCHLD" << endl;
        abort();
    }
    
    if (sigaction(SIGUSR1, &newsignal, NULL) == -1)
    {
        cerr << "sigaction SIGUSR1" << endl;
        abort();
    }
    if (sigaction(SIGUSR2, &newsignal, NULL) == -1)
    {
        cerr << "sigaction SIGUSR2" << endl;
        abort();
    }
    if (sigaction(SIGCHLD, &newsignal, NULL) == -1)
    {
        cerr << "sigaction SIGCHLD" << endl;
        abort();
    }
    
    cout << "manager " << getpid() << ": signal init OK!" << endl;
    managerTCPserver(); //may abort()
}
